package org.yuanzheng.tableAndSql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.types.Row
import org.yuanzheng.source.StationLog

/*TableApi和SQL混用*/
object TestSql2 {
  def main(args: Array[String]): Unit = {
    //创建使用flink原生
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    //隐式转换
    import org.apache.flink.streaming.api.scala._

    //读取数据
    val stream: DataStream[StationLog] = streamEnv.readTextFile(getClass.getResource("/station.log").getPath)
      .map(row => {
        var split = row.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })

    //混用tableApi和SQL
    val table: Table = tableEnv.fromDataStream(stream)
    val result: Table = tableEnv.sqlQuery(s"select sid,sum(duration) as dSun from $table where callType='success' group by sid")
    tableEnv.toRetractStream[Row](result).filter(_._1 == true).print()
    tableEnv.execute("dSum")
  }
}
